package com.share.bigdata.flink.countword.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.share.bigdata.flink.countword.sources.JsonStringSourceFunction;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.Iterator;

public class EventWindowsCountWordApp {
    private static String DATA_TIME_KEY = "dateTime";
    private static String GROUP_KEY = "productCode";

    public static void main(String[] args) {
        try {
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

            //设置事件时间
            env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
            env.getConfig().setAutoWatermarkInterval(10);
            env.setParallelism(1);//默认是使用系统核数个

            // source
            DataStream<String> stream = env.addSource(new JsonStringSourceFunction());
//            DataStream<String> stream = env.addSource(CommonConf.getKafkaConsumerSource());
            System.out.println(EventWindowsCountWordApp.class.getName() + "---开始接收数据");
            stream.print("kafka");

            //数据过滤
            DataStream<JSONObject> jsonStream = stream.map(JSON::parseObject)
                    .filter(e -> StringUtils.isNotBlank(e.getString(DATA_TIME_KEY)));


            //设置拨动时间与提取时间的字段
            SingleOutputStreamOperator<JSONObject> markStream = jsonStream.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<JSONObject>(Time.seconds(3)) {
                @Override
                public long extractTimestamp(JSONObject jsonObject) {
                    return jsonObject.getLong(DATA_TIME_KEY);
                }
            });

            //按照指定字段分组
            KeyedStream<JSONObject, String> groupStream = markStream.keyBy(new KeySelector<JSONObject, String>() {
                @Override
                public String getKey(JSONObject jsonObject) throws Exception {
                    return jsonObject.getString(GROUP_KEY);
                }
            });
//            groupStream.print("group");

            //设定窗口
            //滚动窗口（Tumbling Windows）
            //滑动窗口（Sliding Windows）
//            WindowedStream<JSONObject, String, TimeWindow> windowStream = groupStream.window(TumblingEventTimeWindows.of(Time.seconds(5)));
            WindowedStream<JSONObject, String, TimeWindow> windowStream = groupStream.window(SlidingEventTimeWindows.of(Time.seconds(6), Time.seconds(2)));

            //对窗口求和---到了时间间隔就会计算
            SingleOutputStreamOperator<Object> applyStream = windowStream.apply(
                    new WindowFunction<JSONObject, Object, String, TimeWindow>() {
                        @Override
                        public void apply(String s, TimeWindow timeWindow, Iterable<JSONObject> iterable, Collector<Object> collector) throws Exception {
                            Iterator<JSONObject> iterator = iterable.iterator();
                            int sum = 0;
                            while (iterator.hasNext()) {
                                JSONObject next = iterator.next();
                                sum++;
                            }
                            collector.collect(s + " " + sum);
                            System.out.println("key=" + s + ",start=" + timeWindow.getStart() + ",end" + timeWindow.getEnd());
                        }
                    }
            );
            applyStream.print("applyStream");

            // 将结果写入redis
            // sumStream.addSink(new RedisSink(CommonConf.getFlinkJedisPoolConfig(), new HashSetRedisMapper()));

            // 执行任务
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private static DataStream<String> filter(DataStream<String> source) {
        return source.filter(new FilterFunction<String>() {
            @Override
            public boolean filter(String value) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                Long timeSpace = jsonObject.getLong("transJnlNo");
                if (timeSpace == null) {
                    return false;
                } else {
                    return true;
                }
            }
        });
    }
}
